package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
//import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
//import java.util.Map.Entry;



//import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 根据检测IP作为源IP时的流量统计
 * 2016/02/23 新建
 * @author Administrator
 *
 */
public class BaseStatSumSourceBolt extends AbstractBolt{


	
	/**
	 * 
	 */
	private static final long serialVersionUID = -7271605440932635100L;

	HashMap<String,Integer> sourceIPPV = new HashMap<String,Integer>();
	
	long totalPV = 0L;
	


	
	private void countInLocal(Tuple tuple) {
		
		String sourceip = tuple.getStringByField("sip");
		String idc = tuple.getStringByField("idcno");
		String key = idc + "|" + sourceip;

		if(sourceIPPV.containsKey(key))
		{
			int dipPV = 0;
			dipPV = sourceIPPV.get(key) + 1;
			sourceIPPV.put(key, dipPV);
		}
		else
		{
			sourceIPPV.put(key, 1);
		}
		
		totalPV++;
	}
	
	private void emitCountingData() {
		
		
		Date date = new Date();
//		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
//		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		//SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH");
		SimpleDateFormat sdf_min  = new SimpleDateFormat("mm");
		String dateStr_min = sdf_min.format(date);
		
		String dateStr_h = sdf_h.format(date);
	
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
//		SiteEvaluateMain evaluate = new SiteEvaluateMain(zooserver,logserver);
		
		for(String dkey : sourceIPPV.keySet()) {
			
//			if(mIPs.size() > 0 && !mIPs.containsKey(dkey))
//			{
//				//_collector.ack(input);
//				continue;
//			}
			
			int avgPvalue = sourceIPPV.get(dkey);
			
			String idc = dkey.split("\\|",-1)[0];
			String sourceIP = dkey.split("\\|",-1)[1];
			
			//######  5Min  ##### 
			int nMin5 = (Integer.parseInt(dateStr_min) / 5) * 5;
			String sMin5 = String.format("%02d", nMin5);   
			String dateStr5 = dateStr_h + sMin5;
			String rowkey5 = String.format("%s_%s", dkey,dateStr5);
			String cf = "cf";
			String tbName5 = ApolloProperties.SDS_SESSION_STAT_5MIN;
			hbaseInstance.Add(tbName5, rowkey5, cf,  "DESTIP", sourceIP);
			hbaseInstance.Add(tbName5, rowkey5, cf,  "REPORTTIME", dateStr5);
			hbaseInstance.Add(tbName5, rowkey5, cf, "SPV",(long)avgPvalue);
			
			
			//--------------待屏蔽-------------
//			String tbNametmp = ApolloProperties.SDS_SESSION_STAT_H;
//			hbaseInstance.Add(tbNametmp, rowkey5, cf,  "DESTIP", sourceIP);
//			hbaseInstance.Add(tbNametmp, rowkey5, cf,  "REPORTTIME", dateStr5);
//			hbaseInstance.Add(tbNametmp, rowkey5, cf, "SPV",(long)avgPvalue);
			//--------------待屏蔽-------------
			
			
			
			collector.emit("DELAY",new Values(idc,sourceIP, "SPV", avgPvalue));
			
			//######  30Min  #####
//			int nMin30 = (Integer.parseInt(dateStr_min) / 30) * 30;
//			String sMin30 = String.format("%02d", nMin30);   
//			String dateStr30 = dateStr_h + sMin30;
//			String rowkey30 = String.format("%s_%s", dkey,dateStr30);
//			String tbName30 = ApolloProperties.SDS_SESSION_STAT_30MIN;
//			hbaseInstance.Add(tbName30, rowkey30, cf,  "DESTIP", dkey);
//			hbaseInstance.Add(tbName30, rowkey30, cf,  "REPORTTIME", dateStr30);
//			hbaseInstance.incerment(tbName30, rowkey30, "cf:SPV", avgPvalue);
//			
//			
//			//######  hour  #####
////			int nHour = (Integer.parseInt(dateStr_min) / 30) * 30;
////			String sMin30 = String.format("%02d", nMin30);   
////			String dateStr30 = dateStr_h + sMin30;
//			String rowkeyHour = String.format("%s_%s00", dkey,dateStr_h);
//			String tbNameHour = ApolloProperties.SDS_SESSION_STAT_HOUR;
//			hbaseInstance.Add(tbNameHour, rowkeyHour, cf,  "DESTIP", dkey);
//			hbaseInstance.Add(tbNameHour, rowkeyHour, cf,  "REPORTTIME", dateStr_h + "00");
//			hbaseInstance.incerment(tbNameHour, rowkeyHour, "cf:SPV", avgPvalue);
//			
//			//######  DAY  #####
////			int nHour = (Integer.parseInt(dateStr_min) / 30) * 30;
////			String sMin30 = String.format("%02d", nMin30);   
////			String dateStr30 = dateStr_h + sMin30;
//			String rowkeyDay = String.format("%s_%s", dkey,dateStr_d);
//			String tbNameDay = ApolloProperties.SDS_SESSION_STAT_DAY;
//			hbaseInstance.Add(tbNameDay, rowkeyDay, cf,  "DESTIP", dkey);
//			hbaseInstance.Add(tbNameDay, rowkeyDay, cf,  "REPORTTIME", dateStr_d);
//			hbaseInstance.incerment(tbNameDay, rowkeyDay, "cf:SPV", avgPvalue);
			
		}
		
		sourceIPPV.clear();
	}
	
	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("BaseStatSumBolt Exception",e);
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declareStream("DELAY", new Fields("IDC","IP","TYPE","PV"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}
	


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
